package com.atguigu.tingshu.common.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

@Service
public class KafkaService {

    private static final Logger logger = LoggerFactory.getLogger(KafkaService.class);

    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void sendMsg(String topic, String msg) {
        this.sendMsg(topic, null, null, msg);
    }

    public void sendMsg(String topic, Integer partition, String key, String msg) {
        CompletableFuture<SendResult> future = kafkaTemplate.send(topic, partition, key, msg);
        future.whenCompleteAsync((record, e) -> {
            if (e != null) {
                e.printStackTrace();
                throw new RuntimeException("发送消息到kafka失败！");
            }
        });
    }
}
